package b.cif.impl

import akka.Done
import com.datastax.driver.core.{BoundStatement, PreparedStatement}
import com.lightbend.lagom.scaladsl.persistence.ReadSideProcessor.ReadSideHandler
import com.lightbend.lagom.scaladsl.persistence.cassandra.{CassandraReadSide, CassandraSession}
import com.lightbend.lagom.scaladsl.persistence.{AggregateEventTag, EventStreamElement, ReadSideProcessor}

import scala.collection.immutable.Seq
import scala.concurrent.{ExecutionContext, Future}

class C客户ReadSide(
  readSide: CassandraReadSide,
  session: CassandraSession)(implicit ec: ExecutionContext) extends ReadSideProcessor[T客户Event]{

  @volatile private var v客户Create: PreparedStatement = _ // initialized in prepare

  def f建表(): Future[Done] = {
    val statement = s"""
       |CREATE TABLE IF NOT EXISTS customer (
       |  no ascii,
       |  id varchar,
       |  idType varchar,
       |  PRIMARY KEY (no)
       |)""".stripMargin
    session.executeCreateTable(statement)
  }

  def fSQL语句: Future[Done] = {
    session.prepare("INSERT INTO customer (no, id, idType) VALUES (?, ?, ?)")
      .map { ps => v客户Create = ps; Done}
  }

  private def f新开客户(event: CC新客户Event): Future[Seq[BoundStatement]] = {
    Future.successful({
      val stmt = v客户Create.bind()
      stmt.setString("no", event.v客户.v客户号)
      stmt.setString("id", event.v客户.v证件.v证件号)
      stmt.setString("idType", event.v客户.v证件.v证件类型.toString)
      Seq(stmt)
    })
  }

  override def aggregateTags(): Set[AggregateEventTag[T客户Event]] = {
    Set(O客户Event.TAG)
  }

  override def buildHandler(): ReadSideHandler[T客户Event] = {
    val b = readSide.builder[T客户Event]("CIF")
    b.setGlobalPrepare(f建表)
    b.setPrepare(tag => fSQL语句)
    b.setEventHandler((c: EventStreamElement[CC新客户Event]) => f新开客户(c.event))
    b.build()
  }
}
